package org.yuanzheng.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, KafkaDeserializationSchema}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
// 3.导入隐式转换
import org.apache.flink.streaming.api.scala._

/**
 * @author yuanzheng
 * @date 2020/6/11-21:47
 */
object KafkaSourceKeyValue {
  def main(args: Array[String]): Unit = {
    // 1.初始化流计算环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 2.修改并行度
    streamEnv.setParallelism(1)

    // 4.连接Kafka数据源
    val props = new Properties()
    props.setProperty("bootstrap.servers", "192.168.1.8:9092")
    props.setProperty("group.id", "test-consumer-group")
    props.setProperty("key.deserializer", classOf[StringDeserializer].getName)
    props.setProperty("value.deserializer", classOf[StringDeserializer].getName)
    props.setProperty("auto.offset.reset", "latest")
    val result = streamEnv.addSource(new FlinkKafkaConsumer[(String, String)]("yuan", new MyKafkaReader, props))
    // 6.
    result.print()
    //7.
    streamEnv.execute()
  }

  class MyKafkaReader extends KafkaDeserializationSchema[(String, String)] {
    override def isEndOfStream(t: (String, String)): Boolean = {
      false
    }

    override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): (String, String) = {
      if (record != null) {
        var key = "null"
        var value = "null"
        if (record.key() != null) {
          key = new String(record.key(), "UTF-8")
        }
        if (record.value() != null) {
          value = new String(record.value(), "UTF-8")
        }
        (key, value)
      } else {
        ("null", "null")
      }
    }

    override def getProducedType: TypeInformation[(String, String)] = {
      createTuple2TypeInformation(createTypeInformation[String], createTypeInformation[String])
    }
  }

}
